<!DOCTYPE HTML>
<html lang="en" class="sidebar-visible no-js light">
    <head>
        <!-- Book generated using mdBook -->
        <meta charset="UTF-8">
        <title>使用Flink-SQL实现端到端的流式应用 - 尚硅谷Flink教程</title>
        
        


        <!-- Custom HTML head -->
        


        <meta content="text/html; charset=utf-8" http-equiv="Content-Type">
        <meta name="description" content="">
        <meta name="viewport" content="width=device-width, initial-scale=1">
        <meta name="theme-color" content="#ffffff" />

        
        <link rel="icon" href="favicon.svg">
        
        
        <link rel="shortcut icon" href="favicon.png">
        
        <link rel="stylesheet" href="css/variables.css">
        <link rel="stylesheet" href="css/general.css">
        <link rel="stylesheet" href="css/chrome.css">
        <link rel="stylesheet" href="css/print.css" media="print">

        <!-- Fonts -->
        <link rel="stylesheet" href="FontAwesome/css/font-awesome.css">
        
        <link rel="stylesheet" href="fonts/fonts.css">
        

        <!-- Highlight.js Stylesheets -->
        <link rel="stylesheet" href="highlight.css">
        <link rel="stylesheet" href="tomorrow-night.css">
        <link rel="stylesheet" href="ayu-highlight.css">

        <!-- Custom theme stylesheets -->
        

        
    </head>
    <body>
        <!-- Provide site root to javascript -->
        <script type="text/javascript">
            var path_to_root = "";
            var default_theme = window.matchMedia("(prefers-color-scheme: dark)").matches ? "navy" : "light";
        </script>

        <!-- Work around some values being stored in localStorage wrapped in quotes -->
        <script type="text/javascript">
            try {
                var theme = localStorage.getItem('mdbook-theme');
                var sidebar = localStorage.getItem('mdbook-sidebar');

                if (theme.startsWith('"') && theme.endsWith('"')) {
                    localStorage.setItem('mdbook-theme', theme.slice(1, theme.length - 1));
                }

                if (sidebar.startsWith('"') && sidebar.endsWith('"')) {
                    localStorage.setItem('mdbook-sidebar', sidebar.slice(1, sidebar.length - 1));
                }
            } catch (e) { }
        </script>

        <!-- Set the theme before any content is loaded, prevents flash -->
        <script type="text/javascript">
            var theme;
            try { theme = localStorage.getItem('mdbook-theme'); } catch(e) { }
            if (theme === null || theme === undefined) { theme = default_theme; }
            var html = document.querySelector('html');
            html.classList.remove('no-js')
            html.classList.remove('light')
            html.classList.add(theme);
            html.classList.add('js');
        </script>

        <!-- Hide / unhide sidebar before it is displayed -->
        <script type="text/javascript">
            var html = document.querySelector('html');
            var sidebar = 'hidden';
            if (document.body.clientWidth >= 1080) {
                try { sidebar = localStorage.getItem('mdbook-sidebar'); } catch(e) { }
                sidebar = sidebar || 'visible';
            }
            html.classList.remove('sidebar-visible');
            html.classList.add("sidebar-" + sidebar);
        </script>

        <nav id="sidebar" class="sidebar" aria-label="Table of contents">
            <div class="sidebar-scrollbox">
                <ol class="chapter"><li class="chapter-item expanded "><a href="chapter01-00-00-第一章，有状态的流式处理简介.html"><strong aria-hidden="true">1.</strong> 第一章，有状态的流式处理简介</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter01-01-00-传统数据处理架构.html"><strong aria-hidden="true">1.1.</strong> 传统数据处理架构</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter01-01-01-事务处理.html"><strong aria-hidden="true">1.1.1.</strong> 事务处理</a></li><li class="chapter-item expanded "><a href="chapter01-01-02-分析处理.html"><strong aria-hidden="true">1.1.2.</strong> 分析处理</a></li></ol></li><li class="chapter-item expanded "><a href="chapter01-02-00-有状态的流式处理.html"><strong aria-hidden="true">1.2.</strong> 有状态的流式处理</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter01-02-01-事件驱动应用程序.html"><strong aria-hidden="true">1.2.1.</strong> 事件驱动应用程序</a></li><li class="chapter-item expanded "><a href="chapter01-02-02-数据管道.html"><strong aria-hidden="true">1.2.2.</strong> 数据管道</a></li><li class="chapter-item expanded "><a href="chapter01-02-03-流分析.html"><strong aria-hidden="true">1.2.3.</strong> 流分析</a></li></ol></li><li class="chapter-item expanded "><a href="chapter01-03-00-开源流处理的演进.html"><strong aria-hidden="true">1.3.</strong> 开源流处理的演进</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter01-03-01-流处理的历史.html"><strong aria-hidden="true">1.3.1.</strong> 流处理的历史</a></li></ol></li><li class="chapter-item expanded "><a href="chapter01-04-00-Flink简介.html"><strong aria-hidden="true">1.4.</strong> Flink简介</a></li></ol></li><li class="chapter-item expanded "><a href="chapter02-00-00-第二章，流处理基础.html"><strong aria-hidden="true">2.</strong> 第二章，流处理基础</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter02-01-00-数据流编程简介.html"><strong aria-hidden="true">2.1.</strong> 数据流编程简介</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter02-01-01-数据流图.html"><strong aria-hidden="true">2.1.1.</strong> 数据流图</a></li><li class="chapter-item expanded "><a href="chapter02-01-02-数据并行和任务并行.html"><strong aria-hidden="true">2.1.2.</strong> 数据并行和任务并行</a></li><li class="chapter-item expanded "><a href="chapter02-01-03-数据交换策略.html"><strong aria-hidden="true">2.1.3.</strong> 数据交换策略</a></li></ol></li><li class="chapter-item expanded "><a href="chapter02-02-00-并行处理流数据.html"><strong aria-hidden="true">2.2.</strong> 并行处理流数据</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter02-02-01-延迟和吞吐量.html"><strong aria-hidden="true">2.2.1.</strong> 延迟和吞吐量</a></li><li class="chapter-item expanded "><a href="chapter02-02-02-延迟.html"><strong aria-hidden="true">2.2.2.</strong> 延迟</a></li><li class="chapter-item expanded "><a href="chapter02-02-03-吞吐量.html"><strong aria-hidden="true">2.2.3.</strong> 吞吐量</a></li><li class="chapter-item expanded "><a href="chapter02-02-04-延迟与吞吐量的对比.html"><strong aria-hidden="true">2.2.4.</strong> 延迟与吞吐量的对比</a></li></ol></li><li class="chapter-item expanded "><a href="chapter02-03-00-数据流上的操作.html"><strong aria-hidden="true">2.3.</strong> 数据流上的操作</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter02-03-01-数据摄入和数据吞吐量.html"><strong aria-hidden="true">2.3.1.</strong> 数据摄入和数据吞吐量</a></li><li class="chapter-item expanded "><a href="chapter02-03-02-转换算子.html"><strong aria-hidden="true">2.3.2.</strong> 转换算子</a></li><li class="chapter-item expanded "><a href="chapter02-03-03-滚动聚合.html"><strong aria-hidden="true">2.3.3.</strong> 滚动聚合</a></li><li class="chapter-item expanded "><a href="chapter02-03-04-窗口操作符.html"><strong aria-hidden="true">2.3.4.</strong> 窗口操作符</a></li></ol></li><li class="chapter-item expanded "><a href="chapter02-04-00-时间语义.html"><strong aria-hidden="true">2.4.</strong> 时间语义</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter02-04-01-在流处理中一分钟代表什么？.html"><strong aria-hidden="true">2.4.1.</strong> 在流处理中一分钟代表什么？</a></li><li class="chapter-item expanded "><a href="chapter02-04-02-处理时间.html"><strong aria-hidden="true">2.4.2.</strong> 处理时间</a></li><li class="chapter-item expanded "><a href="chapter02-04-03-事件时间.html"><strong aria-hidden="true">2.4.3.</strong> 事件时间</a></li><li class="chapter-item expanded "><a href="chapter02-04-04-水位线.html"><strong aria-hidden="true">2.4.4.</strong> 水位线</a></li><li class="chapter-item expanded "><a href="chapter02-04-05-处理时间和事件时间.html"><strong aria-hidden="true">2.4.5.</strong> 处理时间和事件时间</a></li></ol></li><li class="chapter-item expanded "><a href="chapter02-05-00-状态和持久化模型.html"><strong aria-hidden="true">2.5.</strong> 状态和持久化模型</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter02-05-01-任务失败.html"><strong aria-hidden="true">2.5.1.</strong> 任务失败</a></li></ol></li></ol></li><li class="chapter-item expanded "><a href="chapter03-00-00-第三章，Flink运行架构.html"><strong aria-hidden="true">3.</strong> 第三章，Flink运行架构</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter03-01-00-系统架构.html"><strong aria-hidden="true">3.1.</strong> 系统架构</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter03-01-01-Flink运行时组件.html"><strong aria-hidden="true">3.1.1.</strong> Flink运行时组件</a></li><li class="chapter-item expanded "><a href="chapter03-01-02-应用部署.html"><strong aria-hidden="true">3.1.2.</strong> 应用部署</a></li><li class="chapter-item expanded "><a href="chapter03-01-03-任务执行.html"><strong aria-hidden="true">3.1.3.</strong> 任务执行</a></li><li class="chapter-item expanded "><a href="chapter03-01-04-高可用配置.html"><strong aria-hidden="true">3.1.4.</strong> 高可用配置</a></li></ol></li><li class="chapter-item expanded "><a href="chapter03-02-Flink中的数据传输.html"><strong aria-hidden="true">3.2.</strong> Flink中的数据传输</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter03-02-01-基于信任度的流控制.html"><strong aria-hidden="true">3.2.1.</strong> 基于信任度的流控制</a></li><li class="chapter-item expanded "><a href="chapter03-02-02-任务链.html"><strong aria-hidden="true">3.2.2.</strong> 任务链</a></li></ol></li><li class="chapter-item expanded "><a href="chapter03-03-00-事件时间处理.html"><strong aria-hidden="true">3.3.</strong> 事件时间处理</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter03-03-01-时间戳.html"><strong aria-hidden="true">3.3.1.</strong> 时间戳</a></li><li class="chapter-item expanded "><a href="chapter03-03-02-水位线.html"><strong aria-hidden="true">3.3.2.</strong> 水位线</a></li><li class="chapter-item expanded "><a href="chapter03-03-03-watermark的传递和事件时间.html"><strong aria-hidden="true">3.3.3.</strong> watermark的传递和事件时间</a></li><li class="chapter-item expanded "><a href="chapter03-03-04-时间戳的分配和水位线的产生.html"><strong aria-hidden="true">3.3.4.</strong> 时间戳的分配和水位线的产生</a></li></ol></li><li class="chapter-item expanded "><a href="chapter03-04-00-状态管理.html"><strong aria-hidden="true">3.4.</strong> 状态管理</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter03-04-01-算子状态.html"><strong aria-hidden="true">3.4.1.</strong> 算子状态</a></li><li class="chapter-item expanded "><a href="chapter03-04-02-键控状态.html"><strong aria-hidden="true">3.4.2.</strong> 键控状态</a></li><li class="chapter-item expanded "><a href="chapter03-04-03-状态后端.html"><strong aria-hidden="true">3.4.3.</strong> 状态后端</a></li><li class="chapter-item expanded "><a href="chapter03-04-04-调整有状态算子的并行度.html"><strong aria-hidden="true">3.4.4.</strong> 调整有状态算子的并行度</a></li></ol></li><li class="chapter-item expanded "><a href="chapter03-05-00-检查点，保存点和状态恢复.html"><strong aria-hidden="true">3.5.</strong> 检查点，保存点和状态恢复</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter03-05-01-一致的检查点.html"><strong aria-hidden="true">3.5.1.</strong> 一致的检查点</a></li><li class="chapter-item expanded "><a href="chapter03-05-02-从一致检查点中恢复状态.html"><strong aria-hidden="true">3.5.2.</strong> 从一致检查点中恢复状态</a></li><li class="chapter-item expanded "><a href="chapter03-05-03-Flink的检查点算法.html"><strong aria-hidden="true">3.5.3.</strong> Flink的检查点算法</a></li><li class="chapter-item expanded "><a href="chapter03-05-04-检查点的性能影响.html"><strong aria-hidden="true">3.5.4.</strong> 检查点的性能影响</a></li><li class="chapter-item expanded "><a href="chapter03-05-05-保存点.html"><strong aria-hidden="true">3.5.5.</strong> 保存点</a></li></ol></li></ol></li><li class="chapter-item expanded "><a href="chapter04-00-00-第四章，编写第一个Flink程序.html"><strong aria-hidden="true">4.</strong> 第四章，编写第一个Flink程序</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter04-01-00-在IDEA中编写Flink程序.html"><strong aria-hidden="true">4.1.</strong> 在IDEA中编写Flink程序</a></li><li class="chapter-item expanded "><a href="chapter04-02-00-下载Flink运行时环境，提交Jar包的运行方式.html"><strong aria-hidden="true">4.2.</strong> 下载Flink运行时环境，提交Jar包的运行方式</a></li></ol></li><li class="chapter-item expanded "><a href="chapter05-00-00-第五章，Flink-DataStream-API.html"><strong aria-hidden="true">5.</strong> 第五章，Flink-DataStream-API</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter05-01-00-你好，Flink！.html"><strong aria-hidden="true">5.1.</strong> 你好，Flink！</a></li><li class="chapter-item expanded "><a href="chapter05-02-00-搭建执行环境.html"><strong aria-hidden="true">5.2.</strong> 搭建执行环境</a></li><li class="chapter-item expanded "><a href="chapter05-03-00-读取输入流.html"><strong aria-hidden="true">5.3.</strong> 读取输入流</a></li><li class="chapter-item expanded "><a href="chapter05-04-00-转换算子的使用.html"><strong aria-hidden="true">5.4.</strong> 转换算子的使用</a></li><li class="chapter-item expanded "><a href="chapter05-05-00-输出结果.html"><strong aria-hidden="true">5.5.</strong> 输出结果</a></li><li class="chapter-item expanded "><a href="chapter05-06-00-执行.html"><strong aria-hidden="true">5.6.</strong> 执行</a></li><li class="chapter-item expanded "><a href="chapter05-07-00-产生传感器读数代码编写.html"><strong aria-hidden="true">5.7.</strong> 产生传感器读数代码编写</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter05-07-01-从批读取数据.html"><strong aria-hidden="true">5.7.1.</strong> 从批读取数据</a></li><li class="chapter-item expanded "><a href="chapter05-07-02-从文件读取数据.html"><strong aria-hidden="true">5.7.2.</strong> 从文件读取数据</a></li><li class="chapter-item expanded "><a href="chapter05-07-03-以Kafka消息队列的数据为数据来源.html"><strong aria-hidden="true">5.7.3.</strong> 以Kafka消息队列的数据为数据来源</a></li><li class="chapter-item expanded "><a href="chapter05-07-04-自定义数据源.html"><strong aria-hidden="true">5.7.4.</strong> 自定义数据源</a></li></ol></li><li class="chapter-item expanded "><a href="chapter05-08-00-转换算子.html"><strong aria-hidden="true">5.8.</strong> 转换算子</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter05-08-01-基本转换算子.html"><strong aria-hidden="true">5.8.1.</strong> 基本转换算子</a></li><li class="chapter-item expanded "><a href="chapter05-08-02-键控流转换算子.html"><strong aria-hidden="true">5.8.2.</strong> 键控流转换算子</a></li><li class="chapter-item expanded "><a href="chapter05-08-03-多流转换算子.html"><strong aria-hidden="true">5.8.3.</strong> 多流转换算子</a></li><li class="chapter-item expanded "><a href="chapter05-08-04-分布式转换算子.html"><strong aria-hidden="true">5.8.4.</strong> 分布式转换算子</a></li></ol></li><li class="chapter-item expanded "><a href="chapter05-09-00-设置并行度.html"><strong aria-hidden="true">5.9.</strong> 设置并行度</a></li><li class="chapter-item expanded "><a href="chapter05-10-00-类型.html"><strong aria-hidden="true">5.10.</strong> 类型</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter05-10-01-支持的数据类型.html"><strong aria-hidden="true">5.10.1.</strong> 支持的数据类型</a></li><li class="chapter-item expanded "><a href="chapter05-10-02-为数据类型创建类型信息.html"><strong aria-hidden="true">5.10.2.</strong> 为数据类型创建类型信息</a></li></ol></li><li class="chapter-item expanded "><a href="chapter05-11-00-定义Key以及引用字段.html"><strong aria-hidden="true">5.11.</strong> 定义Key以及引用字段</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter05-11-01-使用字段位置进行keyBy.html"><strong aria-hidden="true">5.11.1.</strong> 使用字段位置进行keyBy</a></li><li class="chapter-item expanded "><a href="chapter05-11-02-使用字段表达式来进行keyBy.html"><strong aria-hidden="true">5.11.2.</strong> 使用字段表达式来进行keyBy</a></li><li class="chapter-item expanded "><a href="chapter05-11-03-Key选择器.html"><strong aria-hidden="true">5.11.3.</strong> Key选择器</a></li></ol></li><li class="chapter-item expanded "><a href="chapter05-12-00-实现UDF函数，更细粒度的控制流.html"><strong aria-hidden="true">5.12.</strong> 实现UDF函数，更细粒度的控制流</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter05-12-01-函数类.html"><strong aria-hidden="true">5.12.1.</strong> 函数类</a></li><li class="chapter-item expanded "><a href="chapter05-12-02-匿名函数.html"><strong aria-hidden="true">5.12.2.</strong> 匿名函数</a></li><li class="chapter-item expanded "><a href="chapter05-12-03-富函数.html"><strong aria-hidden="true">5.12.3.</strong> 富函数</a></li></ol></li><li class="chapter-item expanded "><a href="chapter05-13-00-Sink.html"><strong aria-hidden="true">5.13.</strong> Sink</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter05-13-01-Kafka.html"><strong aria-hidden="true">5.13.1.</strong> Kafka</a></li><li class="chapter-item expanded "><a href="chapter05-13-02-Redis.html"><strong aria-hidden="true">5.13.2.</strong> Redis</a></li><li class="chapter-item expanded "><a href="chapter05-13-03-ElasticSearch.html"><strong aria-hidden="true">5.13.3.</strong> ElasticSearch</a></li><li class="chapter-item expanded "><a href="chapter05-13-04-JDBC自定义sink.html"><strong aria-hidden="true">5.13.4.</strong> JDBC自定义sink</a></li></ol></li></ol></li><li class="chapter-item expanded "><a href="chapter06-00-00-第六章，基于时间和窗口的操作符.html"><strong aria-hidden="true">6.</strong> 第六章，基于时间和窗口的操作符</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter06-01-00-设置时间属性.html"><strong aria-hidden="true">6.1.</strong> 设置时间属性</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter06-01-01-指定时间戳和产生水位线.html"><strong aria-hidden="true">6.1.1.</strong> 指定时间戳和产生水位线</a></li><li class="chapter-item expanded "><a href="chapter06-01-02-周期性的生成水位线.html"><strong aria-hidden="true">6.1.2.</strong> 周期性的生成水位线</a></li><li class="chapter-item expanded "><a href="chapter06-01-03-如何产生不规则的水位线.html"><strong aria-hidden="true">6.1.3.</strong> 如何产生不规则的水位线</a></li></ol></li><li class="chapter-item expanded "><a href="chapter06-02-00-处理函数.html"><strong aria-hidden="true">6.2.</strong> 处理函数</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter06-02-01-时间服务和定时器.html"><strong aria-hidden="true">6.2.1.</strong> 时间服务和定时器</a></li><li class="chapter-item expanded "><a href="chapter06-02-02-将事件发送到侧输出.html"><strong aria-hidden="true">6.2.2.</strong> 将事件发送到侧输出</a></li><li class="chapter-item expanded "><a href="chapter06-02-03-CoProcessFunction.html"><strong aria-hidden="true">6.2.3.</strong> CoProcessFunction</a></li></ol></li><li class="chapter-item expanded "><a href="chapter06-03-00-窗口操作符.html"><strong aria-hidden="true">6.3.</strong> 窗口操作符</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter06-03-01-定义窗口操作符.html"><strong aria-hidden="true">6.3.1.</strong> 定义窗口操作符</a></li><li class="chapter-item expanded "><a href="chapter06-03-02-内置的窗口分配器.html"><strong aria-hidden="true">6.3.2.</strong> 内置的窗口分配器</a></li><li class="chapter-item expanded "><a href="chapter06-03-03-调用窗口计算函数.html"><strong aria-hidden="true">6.3.3.</strong> 调用窗口计算函数</a></li><li class="chapter-item expanded "><a href="chapter06-03-04-自定义窗口操作符.html"><strong aria-hidden="true">6.3.4.</strong> 自定义窗口操作符</a></li></ol></li><li class="chapter-item expanded "><a href="chapter06-04-00-基于时间的双流Join.html"><strong aria-hidden="true">6.4.</strong> 基于时间的双流Join</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter06-04-01-基于间隔的Join.html"><strong aria-hidden="true">6.4.1.</strong> 基于间隔的Join</a></li><li class="chapter-item expanded "><a href="chapter06-04-02-基于窗口的Join.html"><strong aria-hidden="true">6.4.2.</strong> 基于窗口的Join</a></li></ol></li><li class="chapter-item expanded "><a href="chapter06-05-00-处理迟到的元素.html"><strong aria-hidden="true">6.5.</strong> 处理迟到的元素</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter06-05-01-抛弃迟到元素.html"><strong aria-hidden="true">6.5.1.</strong> 抛弃迟到元素</a></li><li class="chapter-item expanded "><a href="chapter06-05-02-重定向迟到元素.html"><strong aria-hidden="true">6.5.2.</strong> 重定向迟到元素</a></li><li class="chapter-item expanded "><a href="chapter06-05-03-使用迟到元素更新窗口计算结果.html"><strong aria-hidden="true">6.5.3.</strong> 使用迟到元素更新窗口计算结果</a></li></ol></li></ol></li><li class="chapter-item expanded "><a href="chapter07-00-00-第七章，有状态算子和应用.html"><strong aria-hidden="true">7.</strong> 第七章，有状态算子和应用</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter07-01-00-实现有状态的用户自定义函数.html"><strong aria-hidden="true">7.1.</strong> 实现有状态的用户自定义函数</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter07-01-01-在RuntimeContext中定义键控状态.html"><strong aria-hidden="true">7.1.1.</strong> 在RuntimeContext中定义键控状态</a></li><li class="chapter-item expanded "><a href="chapter07-01-02-使用ListCheckpointed接口来实现操作符的列表状态.html"><strong aria-hidden="true">7.1.2.</strong> 使用ListCheckpointed接口来实现操作符的列表状态</a></li><li class="chapter-item expanded "><a href="chapter07-01-03-使用连接的广播状态.html"><strong aria-hidden="true">7.1.3.</strong> 使用连接的广播状态</a></li></ol></li><li class="chapter-item expanded "><a href="chapter07-02-00-配置检查点.html"><strong aria-hidden="true">7.2.</strong> 配置检查点</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter07-02-01-将hdfs配置为状态后端.html"><strong aria-hidden="true">7.2.1.</strong> 将hdfs配置为状态后端</a></li></ol></li><li class="chapter-item expanded "><a href="chapter07-03-保证有状态应用的可维护性.html"><strong aria-hidden="true">7.3.</strong> 保证有状态应用的可维护性</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter07-03-01-指定唯一的操作符标识符.html"><strong aria-hidden="true">7.3.1.</strong> 指定唯一的操作符标识符</a></li><li class="chapter-item expanded "><a href="chapter07-03-02-指定操作符的最大并行度.html"><strong aria-hidden="true">7.3.2.</strong> 指定操作符的最大并行度</a></li></ol></li><li class="chapter-item expanded "><a href="chapter07-04-00-有状态应用的性能和健壮性.html"><strong aria-hidden="true">7.4.</strong> 有状态应用的性能和健壮性</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter07-04-01-选择一个状态后端.html"><strong aria-hidden="true">7.4.1.</strong> 选择一个状态后端</a></li><li class="chapter-item expanded "><a href="chapter07-04-02-防止状态泄露.html"><strong aria-hidden="true">7.4.2.</strong> 防止状态泄露</a></li></ol></li></ol></li><li class="chapter-item expanded "><a href="chapter08-00-00-第八章，读写外部系统.html"><strong aria-hidden="true">8.</strong> 第八章，读写外部系统</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter08-01-00-应用的一致性保证.html"><strong aria-hidden="true">8.1.</strong> 应用的一致性保证</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter08-01-01-幂等性写入.html"><strong aria-hidden="true">8.1.1.</strong> 幂等性写入</a></li><li class="chapter-item expanded "><a href="chapter08-01-02-事务性写入.html"><strong aria-hidden="true">8.1.2.</strong> 事务性写入</a></li></ol></li><li class="chapter-item expanded "><a href="chapter08-02-00-Flink提供的连接器.html"><strong aria-hidden="true">8.2.</strong> Flink提供的连接器</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter08-02-01-Apache-Kafka-Source连接器.html"><strong aria-hidden="true">8.2.1.</strong> Apache-Kafka-Source连接器</a></li><li class="chapter-item expanded "><a href="chapter08-02-02-Apache-Kafka-Sink连接器.html"><strong aria-hidden="true">8.2.2.</strong> Apache-Kafka-Sink连接器</a></li><li class="chapter-item expanded "><a href="chapter08-02-03-Kakfa-Sink的at-least-once保证.html"><strong aria-hidden="true">8.2.3.</strong> Kakfa-Sink的at-least-once保证</a></li><li class="chapter-item expanded "><a href="chapter08-02-04-Kafka-Sink的恰好处理一次语义保证.html"><strong aria-hidden="true">8.2.4.</strong> Kafka-Sink的恰好处理一次语义保证</a></li><li class="chapter-item expanded "><a href="chapter08-02-05-文件系统source连接器.html"><strong aria-hidden="true">8.2.5.</strong> 文件系统source连接器</a></li><li class="chapter-item expanded "><a href="chapter08-02-06-文件系统sink连接器.html"><strong aria-hidden="true">8.2.6.</strong> 文件系统sink连接器</a></li></ol></li><li class="chapter-item expanded "><a href="chapter08-03-00-实现自定义源函数.html"><strong aria-hidden="true">8.3.</strong> 实现自定义源函数</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter08-03-01-可重置的源函数.html"><strong aria-hidden="true">8.3.1.</strong> 可重置的源函数</a></li></ol></li><li class="chapter-item expanded "><a href="chapter08-04-00-实现自定义sink函数.html"><strong aria-hidden="true">8.4.</strong> 实现自定义sink函数</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter08-04-01-幂等性sink连接器.html"><strong aria-hidden="true">8.4.1.</strong> 幂等性sink连接器</a></li><li class="chapter-item expanded "><a href="chapter08-04-02-事务性sink连接器.html"><strong aria-hidden="true">8.4.2.</strong> 事务性sink连接器</a></li></ol></li></ol></li><li class="chapter-item expanded "><a href="chapter09-00-00-第九章，搭建Flink运行流式应用.html"><strong aria-hidden="true">9.</strong> 第九章，搭建Flink运行流式应用</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter09-01-00-部署方式.html"><strong aria-hidden="true">9.1.</strong> 部署方式</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter09-01-01-独立集群.html"><strong aria-hidden="true">9.1.1.</strong> 独立集群</a></li><li class="chapter-item expanded "><a href="chapter09-01-02-Apache-Hadoop-Yarn.html"><strong aria-hidden="true">9.1.2.</strong> Apache-Hadoop-Yarn</a></li></ol></li><li class="chapter-item expanded "><a href="chapter09-02-00-高可用配置.html"><strong aria-hidden="true">9.2.</strong> 高可用配置</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter09-02-01-独立集群高可用配置.html"><strong aria-hidden="true">9.2.1.</strong> 独立集群高可用配置</a></li><li class="chapter-item expanded "><a href="chapter09-02-02-yarn集群高可用配置.html"><strong aria-hidden="true">9.2.2.</strong> yarn集群高可用配置</a></li></ol></li><li class="chapter-item expanded "><a href="chapter09-03-00-与Hadoop集成.html"><strong aria-hidden="true">9.3.</strong> 与Hadoop集成</a></li><li class="chapter-item expanded "><a href="chapter09-04-00-保存点操作.html"><strong aria-hidden="true">9.4.</strong> 保存点操作</a></li><li class="chapter-item expanded "><a href="chapter09-05-00-取消一个应用.html"><strong aria-hidden="true">9.5.</strong> 取消一个应用</a></li><li class="chapter-item expanded "><a href="chapter09-06-00-从保存点启动应用程序.html"><strong aria-hidden="true">9.6.</strong> 从保存点启动应用程序</a></li><li class="chapter-item expanded "><a href="chapter09-07-00-扩容，改变并行度操作.html"><strong aria-hidden="true">9.7.</strong> 扩容，改变并行度操作</a></li></ol></li><li class="chapter-item expanded "><a href="chapter10-00-00-第十章，Flink和流式应用运维.html"><strong aria-hidden="true">10.</strong> 第十章，Flink和流式应用运维</a></li><li class="chapter-item expanded "><a href="chapter11-00-00-第十一章，Flink-CEP简介.html"><strong aria-hidden="true">11.</strong> 第十一章，Flink-CEP简介</a></li><li class="chapter-item expanded "><a href="chapter12-00-00-第十二章，Table-API和Flink-SQL.html"><strong aria-hidden="true">12.</strong> 第十二章，Table-API和Flink-SQL</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter12-01-00-整体介绍.html"><strong aria-hidden="true">12.1.</strong> 整体介绍</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter12-01-01-什么是Table-API和Flink-SQL.html"><strong aria-hidden="true">12.1.1.</strong> 什么是Table-API和Flink-SQL</a></li><li class="chapter-item expanded "><a href="chapter12-01-02-需要引入的依赖.html"><strong aria-hidden="true">12.1.2.</strong> 需要引入的依赖</a></li><li class="chapter-item expanded "><a href="chapter12-01-03-两种planner（old-&-blink）的区别.html"><strong aria-hidden="true">12.1.3.</strong> 两种planner（old-&amp;-blink）的区别</a></li></ol></li><li class="chapter-item expanded "><a href="chapter12-02-00-API调用.html"><strong aria-hidden="true">12.2.</strong> API调用</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter12-02-01-基本程序结构.html"><strong aria-hidden="true">12.2.1.</strong> 基本程序结构</a></li><li class="chapter-item expanded "><a href="chapter12-02-02-创建表环境.html"><strong aria-hidden="true">12.2.2.</strong> 创建表环境</a></li><li class="chapter-item expanded "><a href="chapter12-02-03-在Catalog中注册表.html"><strong aria-hidden="true">12.2.3.</strong> 在Catalog中注册表</a></li><li class="chapter-item expanded "><a href="chapter12-02-04-表的查询.html"><strong aria-hidden="true">12.2.4.</strong> 表的查询</a></li><li class="chapter-item expanded "><a href="chapter12-02-05-将DataStream转换成表.html"><strong aria-hidden="true">12.2.5.</strong> 将DataStream转换成表</a></li><li class="chapter-item expanded "><a href="chapter12-02-06-创建临时视图.html"><strong aria-hidden="true">12.2.6.</strong> 创建临时视图</a></li><li class="chapter-item expanded "><a href="chapter12-02-07-输出表.html"><strong aria-hidden="true">12.2.7.</strong> 输出表</a></li><li class="chapter-item expanded "><a href="chapter12-02-08-将表转换成DataStream.html"><strong aria-hidden="true">12.2.8.</strong> 将表转换成DataStream</a></li><li class="chapter-item expanded "><a href="chapter12-02-09-Query的解释和执行.html"><strong aria-hidden="true">12.2.9.</strong> Query的解释和执行</a></li></ol></li><li class="chapter-item expanded "><a href="chapter12-03-00-流处理中的特殊概念.html"><strong aria-hidden="true">12.3.</strong> 流处理中的特殊概念</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter12-03-01-流处理和关系代数（表，及SQL）的区别.html"><strong aria-hidden="true">12.3.1.</strong> 流处理和关系代数（表，及SQL）的区别</a></li><li class="chapter-item expanded "><a href="chapter12-03-02-动态表.html"><strong aria-hidden="true">12.3.2.</strong> 动态表</a></li><li class="chapter-item expanded "><a href="chapter12-03-03-流式持续查询的过程.html"><strong aria-hidden="true">12.3.3.</strong> 流式持续查询的过程</a></li><li class="chapter-item expanded "><a href="chapter12-03-04-时间特性.html"><strong aria-hidden="true">12.3.4.</strong> 时间特性</a></li></ol></li><li class="chapter-item expanded "><a href="chapter12-04-00-窗口.html"><strong aria-hidden="true">12.4.</strong> 窗口</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter12-04-01-分组窗口.html"><strong aria-hidden="true">12.4.1.</strong> 分组窗口</a></li><li class="chapter-item expanded "><a href="chapter12-04-02-Over-Windows.html"><strong aria-hidden="true">12.4.2.</strong> Over-Windows</a></li><li class="chapter-item expanded "><a href="chapter12-04-03-SQL中窗口的定义.html"><strong aria-hidden="true">12.4.3.</strong> SQL中窗口的定义</a></li><li class="chapter-item expanded "><a href="chapter12-04-04-代码练习（以分组滚动窗口为例）.html"><strong aria-hidden="true">12.4.4.</strong> 代码练习（以分组滚动窗口为例）</a></li></ol></li><li class="chapter-item expanded "><a href="chapter12-05-00-函数.html"><strong aria-hidden="true">12.5.</strong> 函数</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter12-05-01-系统内置函数.html"><strong aria-hidden="true">12.5.1.</strong> 系统内置函数</a></li><li class="chapter-item expanded "><a href="chapter12-05-02-UDF.html"><strong aria-hidden="true">12.5.2.</strong> UDF</a></li></ol></li><li class="chapter-item expanded "><a href="chapter12-06-00-Flink和Hive集成.html"><strong aria-hidden="true">12.6.</strong> Flink与Hive集成</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter12-06-01-Maven依赖.html"><strong aria-hidden="true">12.6.1.</strong> Maven依赖</a></li><li class="chapter-item expanded "><a href="chapter12-06-02-示例程序.html"><strong aria-hidden="true">12.6.2.</strong> 示例程序</a></li><li class="chapter-item expanded "><a href="chapter12-06-03-一个复杂一点的程序.html"><strong aria-hidden="true">12.6.3.</strong> 一个复杂一点的程序</a></li><li class="chapter-item expanded "><a href="chapter12-06-04-彻底重置hadoop和hive的方法.html"><strong aria-hidden="true">12.6.4.</strong> 彻底重置hadoop和hive的方法</a></li></ol></li></ol></li><li class="chapter-item expanded "><a href="chapter13-00-00-第十三章，尚硅谷大数据技术之电商用户行为分析.html"><strong aria-hidden="true">13.</strong> 第十三章，尚硅谷大数据技术之电商用户行为分析</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter13-01-00-数据集解析.html"><strong aria-hidden="true">13.1.</strong> 数据集解析</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter13-01-01-淘宝数据集解析.html"><strong aria-hidden="true">13.1.1.</strong> 淘宝数据集解析</a></li><li class="chapter-item expanded "><a href="chapter13-01-02-Apache服务器日志数据集解析.html"><strong aria-hidden="true">13.1.2.</strong> Apache服务器日志数据集解析</a></li></ol></li><li class="chapter-item expanded "><a href="chapter13-02-00-实时热门商品统计.html"><strong aria-hidden="true">13.2.</strong> 实时热门商品统计</a></li><li class="chapter-item expanded "><a href="chapter13-03-00-实时流量统计.html"><strong aria-hidden="true">13.3.</strong> 实时流量统计</a></li><li class="chapter-item expanded "><a href="chapter13-04-00-Uv统计的布隆过滤器实现.html"><strong aria-hidden="true">13.4.</strong> Uv统计的布隆过滤器实现</a></li><li class="chapter-item expanded "><a href="chapter13-05-00-APP分渠道数据统计.html"><strong aria-hidden="true">13.5.</strong> APP分渠道数据统计</a></li><li class="chapter-item expanded "><a href="chapter13-06-00-APP不分渠道数据统计.html"><strong aria-hidden="true">13.6.</strong> APP不分渠道数据统计</a></li><li class="chapter-item expanded "><a href="chapter13-07-00-恶意登陆实现.html"><strong aria-hidden="true">13.7.</strong> 恶意登陆实现</a></li><li class="chapter-item expanded "><a href="chapter13-08-00-订单支付实时监控.html"><strong aria-hidden="true">13.8.</strong> 订单支付实时监控</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter13-08-01-使用Flink-CEP来实现.html"><strong aria-hidden="true">13.8.1.</strong> 使用Flink-CEP来实现</a></li><li class="chapter-item expanded "><a href="chapter13-08-02-使用Process-Function实现订单超时需求.html"><strong aria-hidden="true">13.8.2.</strong> 使用Process-Function实现订单超时需求</a></li></ol></li><li class="chapter-item expanded "><a href="chapter13-09-00-实时对帐：实现两条流的Join.html"><strong aria-hidden="true">13.9.</strong> 实时对帐：实现两条流的Join</a></li><li class="chapter-item expanded "><a href="chapter13-10-00-使用Flink-SQL实现实时热门商品统计.html"><strong aria-hidden="true">13.10.</strong> 使用Flink-SQL实现实时热门商品统计</a></li><li class="chapter-item expanded "><a href="chapter13-11-00-使用Flink-SQL实现端到端的流式应用.html" class="active"><strong aria-hidden="true">13.11.</strong> 使用Flink-SQL实现端到端的流式应用</a></li></ol></li><li class="chapter-item expanded "><a href="chapter14-00-第十四章，常见面试题解答.html"><strong aria-hidden="true">14.</strong> 第十四章，常见面试题解答</a></li><li><ol class="section"><li class="chapter-item expanded "><a href="chapter14-01-面试题一.html"><strong aria-hidden="true">14.1.</strong> 面试题一</a></li><li class="chapter-item expanded "><a href="chapter14-02-面试题二.html"><strong aria-hidden="true">14.2.</strong> 面试题二</a></li><li class="chapter-item expanded "><a href="chapter14-03-面试题三.html"><strong aria-hidden="true">14.3.</strong> 面试题三</a></li><li class="chapter-item expanded "><a href="chapter14-04-面试题四.html"><strong aria-hidden="true">14.4.</strong> 面试题四</a></li><li class="chapter-item expanded "><a href="chapter14-05-面试题五.html"><strong aria-hidden="true">14.5.</strong> 面试题五</a></li><li class="chapter-item expanded "><a href="chapter14-06-面试题六.html"><strong aria-hidden="true">14.6.</strong> 面试题六</a></li><li class="chapter-item expanded "><a href="chapter14-07-面试题七.html"><strong aria-hidden="true">14.7.</strong> 面试题七</a></li><li class="chapter-item expanded "><a href="chapter14-08-面试题八.html"><strong aria-hidden="true">14.8.</strong> 面试题八</a></li><li class="chapter-item expanded "><a href="chapter14-09-面试题九.html"><strong aria-hidden="true">14.9.</strong> 面试题九</a></li><li class="chapter-item expanded "><a href="chapter14-10-面试题十.html"><strong aria-hidden="true">14.10.</strong> 面试题十</a></li><li class="chapter-item expanded "><a href="chapter14-11-面试题十一.html"><strong aria-hidden="true">14.11.</strong> 面试题十一</a></li><li class="chapter-item expanded "><a href="chapter14-12-面试题十二.html"><strong aria-hidden="true">14.12.</strong> 面试题十二</a></li><li class="chapter-item expanded "><a href="chapter14-13-面试题十三.html"><strong aria-hidden="true">14.13.</strong> 面试题十三</a></li><li class="chapter-item expanded "><a href="chapter14-14-面试题十四.html"><strong aria-hidden="true">14.14.</strong> 面试题十四</a></li><li class="chapter-item expanded "><a href="chapter14-15-面试题十五.html"><strong aria-hidden="true">14.15.</strong> 面试题十五</a></li><li class="chapter-item expanded "><a href="chapter14-16-面试题十六.html"><strong aria-hidden="true">14.16.</strong> 面试题十六</a></li><li class="chapter-item expanded "><a href="chapter14-17-面试题十七.html"><strong aria-hidden="true">14.17.</strong> 面试题十七</a></li><li class="chapter-item expanded "><a href="chapter14-18-面试题十八.html"><strong aria-hidden="true">14.18.</strong> 面试题十八</a></li><li class="chapter-item expanded "><a href="chapter14-19-面试题十九.html"><strong aria-hidden="true">14.19.</strong> 面试题十九</a></li><li class="chapter-item expanded "><a href="chapter14-20-面试题二十.html"><strong aria-hidden="true">14.20.</strong> 面试题二十</a></li></ol></li></ol>
            </div>
            <div id="sidebar-resize-handle" class="sidebar-resize-handle"></div>
        </nav>

        <div id="page-wrapper" class="page-wrapper">

            <div class="page">
                
                <div id="menu-bar-hover-placeholder"></div>
                <div id="menu-bar" class="menu-bar sticky bordered">
                    <div class="left-buttons">
                        <button id="sidebar-toggle" class="icon-button" type="button" title="Toggle Table of Contents" aria-label="Toggle Table of Contents" aria-controls="sidebar">
                            <i class="fa fa-bars"></i>
                        </button>
                        <button id="theme-toggle" class="icon-button" type="button" title="Change theme" aria-label="Change theme" aria-haspopup="true" aria-expanded="false" aria-controls="theme-list">
                            <i class="fa fa-paint-brush"></i>
                        </button>
                        <ul id="theme-list" class="theme-popup" aria-label="Themes" role="menu">
                            <li role="none"><button role="menuitem" class="theme" id="light">Light (default)</button></li>
                            <li role="none"><button role="menuitem" class="theme" id="rust">Rust</button></li>
                            <li role="none"><button role="menuitem" class="theme" id="coal">Coal</button></li>
                            <li role="none"><button role="menuitem" class="theme" id="navy">Navy</button></li>
                            <li role="none"><button role="menuitem" class="theme" id="ayu">Ayu</button></li>
                        </ul>
                        
                        <button id="search-toggle" class="icon-button" type="button" title="Search. (Shortkey: s)" aria-label="Toggle Searchbar" aria-expanded="false" aria-keyshortcuts="S" aria-controls="searchbar">
                            <i class="fa fa-search"></i>
                        </button>
                        
                    </div>

                    <h1 class="menu-title">尚硅谷Flink教程</h1>

                    <div class="right-buttons">
                        <a href="print.html" title="Print this book" aria-label="Print this book">
                            <i id="print-button" class="fa fa-print"></i>
                        </a>
                        
                    </div>
                </div>

                
                <div id="search-wrapper" class="hidden">
                    <form id="searchbar-outer" class="searchbar-outer">
                        <input type="search" name="search" id="searchbar" name="searchbar" placeholder="Search this book ..." aria-controls="searchresults-outer" aria-describedby="searchresults-header">
                    </form>
                    <div id="searchresults-outer" class="searchresults-outer hidden">
                        <div id="searchresults-header" class="searchresults-header"></div>
                        <ul id="searchresults">
                        </ul>
                    </div>
                </div>
                

                <!-- Apply ARIA attributes after the sidebar and the sidebar toggle button are added to the DOM -->
                <script type="text/javascript">
                    document.getElementById('sidebar-toggle').setAttribute('aria-expanded', sidebar === 'visible');
                    document.getElementById('sidebar').setAttribute('aria-hidden', sidebar !== 'visible');
                    Array.from(document.querySelectorAll('#sidebar a')).forEach(function(link) {
                        link.setAttribute('tabIndex', sidebar === 'visible' ? 0 : -1);
                    });
                </script>

                <div id="content" class="content">
                    <main>
                        <h1><a class="header" href="#flink-sql-demo-构建一个端到端的流式应用" id="flink-sql-demo-构建一个端到端的流式应用">Flink SQL Demo: 构建一个端到端的流式应用</a></h1>
<p>本文将基于 Kafka, MySQL, Elasticsearch, Kibana，使用 Flink SQL 构建一个电商用户行为的实时分析应用。本文所有的实战演练都将在 Flink SQL CLI 上执行，全程只涉及 SQL 纯文本，无需一行 Java/Scala 代码，无需安装 IDE。本实战演练的最终效果图：</p>
<p><img src="https://img.alicdn.com/tfs/TB1xc2ewlr0gK0jSZFnXXbRRXXa-3104-1978.png" alt="" /></p>
<!-- more -->
<h2><a class="header" href="#准备" id="准备">准备</a></h2>
<p>一台装有 Docker 的 Linux 或 MacOS 计算机。</p>
<h3><a class="header" href="#使用-docker-compose-启动容器" id="使用-docker-compose-启动容器">使用 Docker Compose 启动容器</a></h3>
<p>本实战演示所依赖的组件全都编排到了容器中，因此可以通过 <code>docker-compose</code> 一键启动。你可以通过 <code>wget</code> 命令自动下载该 <code>docker-compose.yml</code> 文件，也可以手动下载。</p>
<pre><code class="language-bash">mkdir flink-sql-demo; cd flink-sql-demo;
wget https://raw.githubusercontent.com/wuchong/flink-sql-demo/v1.11-CN/docker-compose.yml
</code></pre>
<p>该 Docker Compose 中包含的容器有：</p>
<ul>
<li><strong>Flink SQL Client</strong>: 用于提交 Flink SQL</li>
<li><strong>Flink集群</strong>: 包含一个 JobManager 和 一个 TaskManager 用于运行 SQL 任务。</li>
<li><strong>DataGen:</strong> 数据生成器。容器启动后会自动开始生成用户行为数据，并发送到 Kafka 集群中。默认每秒生成 2000 条数据，能持续生成一个多小时。也可以更改 <code>docker-compose.yml</code> 中 datagen 的 <code>speedup</code> 参数来调整生成速率（重启 docker compose 才能生效）。</li>
<li><strong>MySQL:</strong> 集成了 MySQL 5.7 ，以及预先创建好了类目表（<code>category</code>），预先填入了子类目与顶级类目的映射关系，后续作为维表使用。</li>
<li><strong>Kafka:</strong> 主要用作数据源。DataGen 组件会自动将数据灌入这个容器中。</li>
<li><strong>Zookeeper:</strong> Kafka 容器依赖。</li>
<li><strong>Elasticsearch:</strong> 主要存储 Flink SQL 产出的数据。</li>
<li><strong>Kibana:</strong> 可视化 Elasticsearch 中的数据。</li>
</ul>
<p>在启动容器前，建议修改 Docker 的配置，将资源调整到 4GB 以及 4核。启动所有的容器，只需要在 <code>docker-compose.yml</code> 所在目录下运行如下命令。</p>
<pre><code class="language-bash">docker-compose up -d
</code></pre>
<p>该命令会以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 <code>docker ps</code> 来观察上述的五个容器是否正常启动了。 也可以访问 http://localhost:5601/ 来查看 Kibana 是否运行正常。</p>
<p>另外可以通过如下命令停止所有的容器：</p>
<pre><code class="language-bash">docker-compose down
</code></pre>
<h3><a class="header" href="#进入-sql-cli-客户端" id="进入-sql-cli-客户端">进入 SQL CLI 客户端</a></h3>
<p>运行如下命令进入 SQL CLI 客户端：</p>
<pre><code class="language-bash">docker-compose exec sql-client ./sql-client.sh
</code></pre>
<p>该命令会在容器中启动 SQL CLI 客户端。你应该能在 CLI 客户端中看到如下的环境界面。</p>
<pre><code>
▒▓██▓██▒
▓████▒▒█▓▒▓███▓▒
▓███▓░░        ▒▒▒▓██▒  ▒
░██▒   ▒▒▓▓█▓▓▒░      ▒████
██▒         ░▒▓███▒    ▒█▒█▒
░▓█            ███   ▓░▒██
▓█       ▒▒▒▒▒▓██▓░▒░▓▓█
█░ █   ▒▒░       ███▓▓█ ▒█▒▒▒
████░   ▒▓█▓      ██▒▒▒ ▓███▒
░▒█▓▓██       ▓█▒    ▓█▒▓██▓ ░█░
▓░▒▓████▒ ██         ▒█    █▓░▒█▒░▒█▒
███▓░██▓  ▓█           █   █▓ ▒▓█▓▓█▒
░██▓  ░█░            █  █▒ ▒█████▓▒ ██▓░▒
███░ ░ █░          ▓ ░█ █████▒░░    ░█░▓  ▓░
██▓█ ▒▒▓▒          ▓███████▓░       ▒█▒ ▒▓ ▓██▓
▒██▓ ▓█ █▓█       ░▒█████▓▓▒░         ██▒▒  █ ▒  ▓█▒
▓█▓  ▓█ ██▓ ░▓▓▓▓▓▓▓▒              ▒██▓           ░█▒
▓█    █ ▓███▓▒░              ░▓▓▓███▓          ░▒░ ▓█
██▓    ██▒    ░▒▓▓███▓▓▓▓▓██████▓▒            ▓███  █
▓███▒ ███   ░▓▓▒░░   ░▓████▓░                  ░▒▓▒  █▓
█▓▒▒▓▓██  ░▒▒░░░▒▒▒▒▓██▓░                            █▓
██ ▓░▒█   ▓▓▓▓▒░░  ▒█▓       ▒▓▓██▓    ▓▒          ▒▒▓
▓█▓ ▓▒█  █▓░  ░▒▓▓██▒            ░▓█▒   ▒▒▒░▒▒▓█████▒
██░ ▓█▒█▒  ▒▓▓▒  ▓█                █░      ░░░░   ░█▒
▓█   ▒█▓   ░     █░                ▒█              █▓
█▓   ██         █░                 ▓▓        ▒█▓▓▓▒█░
█▓ ░▓██░       ▓▒                  ▓█▓▒░░░▒▓█░    ▒█
██   ▓█▓░      ▒                    ░▒█▒██▒      ▓▓
▓█▒   ▒█▓▒░                         ▒▒ █▒█▓▒▒░░▒██
░██▒    ▒▓▓▒                     ▓██▓▒█▒ ░▓▓▓▓▒█▓
░▓██▒                          ▓░  ▒█▓█  ░░▒▒▒
▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓  ▓░▒█░

______ _ _       _       _____  ____  _         _____ _ _            _  BETA
|  ____| (_)     | |     / ____|/ __ \| |       / ____| (_)          | |
| |__  | |_ _ __ | | __ | (___ | |  | | |      | |    | |_  ___ _ __ | |_
                               |  __| | | | '_ \| |/ /  \___ \| |  | | |      | |    | | |/ _ \ '_ \| __|
                               | |    | | | | | |   &lt;   ____) | |__| | |____  | |____| | |  __/ | | | |_
|_|    |_|_|_| |_|_|\_\ |_____/ \___\_\______|  \_____|_|_|\___|_| |_|\__|

Welcome! Enter HELP to list all available commands. QUIT to exit.
</code></pre>
<h2><a class="header" href="#使用-ddl-创建-kafka-表" id="使用-ddl-创建-kafka-表">使用 DDL 创建 Kafka 表</a></h2>
<p>Datagen 容器在启动后会往 Kafka 的 <code>user_behavior</code> topic 中持续不断地写入数据。数据包含了2017年11月27日一天的用户行为（行为包括点击、购买、加购、喜欢），每一行表示一条用户行为，以 JSON 的格式由用户ID、商品ID、商品类目ID、行为类型和时间组成。该原始数据集来自<a href="https://tianchi.aliyun.com/dataset/dataDetail?dataId=649">阿里云天池公开数据集</a>，特此鸣谢。</p>
<p>我们可以在 <code>docker-compose.yml</code> 所在目录下运行如下命令，查看 Kafka 集群中生成的前10条数据。</p>
<pre><code class="language-bash">docker-compose exec kafka bash -c 'kafka-console-consumer.sh --topic user_behavior --bootstrap-server kafka:9094 --from-beginning --max-messages 10'
</code></pre>
<pre><code class="language-json">{&quot;user_id&quot;: &quot;952483&quot;, &quot;item_id&quot;:&quot;310884&quot;, &quot;category_id&quot;: &quot;4580532&quot;, &quot;behavior&quot;: &quot;pv&quot;, &quot;ts&quot;: &quot;2017-11-27T00:00:00Z&quot;}
{&quot;user_id&quot;: &quot;794777&quot;, &quot;item_id&quot;:&quot;5119439&quot;, &quot;category_id&quot;: &quot;982926&quot;, &quot;behavior&quot;: &quot;pv&quot;, &quot;ts&quot;: &quot;2017-11-27T00:00:00Z&quot;}
...
</code></pre>
<p>有了数据源后，我们就可以用 DDL 去创建并连接这个 Kafka 中的 topic 了。在 Flink SQL CLI 中执行该 DDL。</p>
<pre><code class="language-sql">CREATE TABLE user_behavior (
                            user_id BIGINT,
                            item_id BIGINT,
                            category_id BIGINT,
                            behavior STRING,
                            ts TIMESTAMP(3),
                            proctime AS PROCTIME(),   -- generates processing-time attribute using computed column
                            WATERMARK FOR ts AS ts - INTERVAL '5' SECOND  -- defines watermark on ts column, marks ts as event-time attribute
                            ) WITH (
                            'connector' = 'kafka',  -- using kafka connector
                            'topic' = 'user_behavior',  -- kafka topic
                            'scan.startup.mode' = 'earliest-offset',  -- reading from the beginning
                            'properties.bootstrap.servers' = 'kafka:9094',  -- kafka broker address
                            'format' = 'json'  -- the data format is json
                            );
</code></pre>
<p>如上我们按照数据的格式声明了 5 个字段，除此之外，我们还通过计算列语法和 <code>PROCTIME()</code> 内置函数声明了一个产生处理时间的虚拟列。我们还通过 WATERMARK 语法，在 ts 字段上声明了 watermark 策略（容忍5秒乱序）， ts 字段因此也成了事件时间列。关于时间属性以及 DDL 语法可以阅读官方文档了解更多：</p>
<ul>
<li>时间属性：
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/streaming/time_attributes.html</li>
<li>DDL：
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/create.html#create-table</li>
</ul>
<p>在 SQL CLI 中成功创建 Kafka 表后，可以通过 <code>show tables;</code> 和 <code>describe user_behavior;</code> 来查看目前已注册的表，以及表的详细信息。我们也可以直接在 SQL CLI 中运行 <code>SELECT * FROM user_behavior;</code> 预览下数据（按<code>q</code>退出）。</p>
<p>接下来，我们会通过三个实战场景来更深入地了解 Flink SQL 。</p>
<h2><a class="header" href="#统计每小时的成交量" id="统计每小时的成交量">统计每小时的成交量</a></h2>
<h3><a class="header" href="#使用-ddl-创建-elasticsearch-表" id="使用-ddl-创建-elasticsearch-表">使用 DDL 创建 Elasticsearch 表</a></h3>
<p>我们先在 SQL CLI 中创建一个 ES 结果表，根据场景需求主要需要保存两个数据：小时、成交量。</p>
<pre><code class="language-sql">CREATE TABLE buy_cnt_per_hour (
                               hour_of_day BIGINT,
                               buy_cnt BIGINT
                               ) WITH (
                               'connector' = 'elasticsearch-7', -- using elasticsearch connector
                               'hosts' = 'http://elasticsearch:9200',  -- elasticsearch address
                               'index' = 'buy_cnt_per_hour'  -- elasticsearch index name, similar to database table name
                               );
</code></pre>
<p>我们不需要在 Elasticsearch 中事先创建 <code>buy_cnt_per_hour</code> 索引，Flink Job 会自动创建该索引。</p>
<h3><a class="header" href="#提交-query" id="提交-query">提交 Query</a></h3>
<p>统计每小时的成交量就是每小时共有多少 &quot;buy&quot; 的用户行为。因此会需要用到 TUMBLE 窗口函数，按照一小时切窗。然后每个窗口分别统计 &quot;buy&quot; 的个数，这可以通过先过滤出 &quot;buy&quot; 的数据，然后 <code>COUNT(*)</code> 实现。</p>
<pre><code class="language-sql">INSERT INTO buy_cnt_per_hour
SELECT HOUR(TUMBLE_START(ts, INTERVAL '1' HOUR)), COUNT(*)
FROM user_behavior
WHERE behavior = 'buy'
GROUP BY TUMBLE(ts, INTERVAL '1' HOUR);
</code></pre>
<p>这里我们使用 <code>HOUR</code> 内置函数，从一个 TIMESTAMP 列中提取出一天中第几个小时的值。使用了 <code>INSERT INTO</code>将 query 的结果持续不断地插入到上文定义的 es 结果表中（可以将 es 结果表理解成 query 的物化视图）。另外可以阅读该文档了解更多关于窗口聚合的内容：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/queries.html#group-windows</p>
<p>在 Flink SQL CLI 中运行上述查询后，在 Flink Web UI 中就能看到提交的任务，该任务是一个流式任务，因此会一直运行。</p>
<p><img src="https://img.alicdn.com/tfs/TB1DU7ovubviK0jSZFNXXaApXXa-2878-1310.png" alt="" /></p>
<p>可以看到凌晨是一天中成交量的低谷。</p>
<h3><a class="header" href="#使用-kibana-可视化结果" id="使用-kibana-可视化结果">使用 Kibana 可视化结果</a></h3>
<p>我们已经通过 Docker Compose 启动了 Kibana 容器，可以通过 http://localhost:5601 访问 Kibana。首先我们需要先配置一个 index pattern。点击左侧工具栏的 &quot;Management&quot;，就能找到 &quot;Index Patterns&quot;。点击 &quot;Create Index Pattern&quot;，然后通过输入完整的索引名 &quot;buy_cnt_per_hour&quot; 创建 index pattern。创建完成后， Kibana 就知道了我们的索引，我们就可以开始探索数据了。</p>
<p>先点击左侧工具栏的&quot;Discovery&quot;按钮，Kibana 就会列出刚刚创建的索引中的内容。</p>
<p><img src="https://img.alicdn.com/tfs/TB1xDYawbY1gK0jSZTEXXXDQVXa-2878-946.png" alt="" /></p>
<p>接下来，我们先创建一个 Dashboard 用来展示各个可视化的视图。点击页面左侧的&quot;Dashboard&quot;，创建一个名为 ”用户行为日志分析“ 的Dashboard。然后点击 &quot;Create New&quot; 创建一个新的视图，选择 &quot;Area&quot; 面积图，选择 &quot;buy_cnt_per_hour&quot; 索引，按照如下截图中的配置（左侧）画出成交量面积图，并保存为”每小时成交量“。</p>
<p><img src="https://img.alicdn.com/tfs/TB19ae.woT1gK0jSZFhXXaAtVXa-2874-1596.png" alt="" /></p>
<h2><a class="header" href="#统计一天每10分钟累计独立用户数" id="统计一天每10分钟累计独立用户数">统计一天每10分钟累计独立用户数</a></h2>
<p>另一个有意思的可视化是统计一天中每一刻的累计独立用户数（uv），也就是每一刻的 uv 数都代表从0点到当前时刻为止的总计 uv 数，因此该曲线肯定是单调递增的。</p>
<p>我们仍然先在 SQL CLI 中创建一个 Elasticsearch 表，用于存储结果汇总数据。主要字段有：日期时间和累积 uv 数。我们将日期时间作为 Elasticsearch 中的 document id，便于更新该日期时间的 uv 值。</p>
<pre><code class="language-sql">CREATE TABLE cumulative_uv (
                            date_str STRING,
                            time_str STRING,
                            uv BIGINT,
                            PRIMARY KEY (date_str, time_str) NOT ENFORCED
                            ) WITH (
                            'connector' = 'elasticsearch-7',
                            'hosts' = 'http://elasticsearch:9200',
                            'index' = 'cumulative_uv'
                            );
</code></pre>
<p>为了实现该曲线，我们先抽取出日期和时间字段，我们使用 <code>DATE_FORMAT</code> 抽取出基本的日期与时间，再用 <code>SUBSTR</code> 和 字符串连接函数 <code>||</code> 将时间修正到10分钟级别，如: <code>12:10</code>, <code>12:20</code>。其次，我们在外层查询上基于日期分组，求当前最大的时间，和 UV，写入到 Elasticsearch 的索引中。UV 的统计我们通过内置的 <code>COUNT(DISTINCT user_id)</code>来完成，Flink SQL 内部对 COUNT DISTINCT 做了非常多的优化，因此可以放心使用。</p>
<p>这里之所以需要求最大的时间，同时又按日期+时间作为主键写入到 Elasticsearch，是因为我们在计算累积 UV 数。</p>
<pre><code class="language-sql">INSERT INTO cumulative_uv
SELECT date_str, MAX(time_str), COUNT(DISTINCT user_id) as uv
FROM (
      SELECT
      DATE_FORMAT(ts, 'yyyy-MM-dd') as date_str,
      SUBSTR(DATE_FORMAT(ts, 'HH:mm'),1,4) || '0' as time_str,
      user_id
      FROM user_behavior)
GROUP BY date_str;
</code></pre>
<p>提交上述查询后，在 Kibana 中创建 <code>cumulative_uv</code> 的 index pattern，然后在 Dashboard 中创建一个&quot;Line&quot;折线图，选择 <code>cumulative_uv</code> 索引，按照如下截图中的配置（左侧）画出累计独立用户数曲线，并保存。</p>
<p><img src="https://img.alicdn.com/tfs/TB1xU5.wkY2gK0jSZFgXXc5OFXa-2878-1598.png" alt="" /></p>
<h2><a class="header" href="#顶级类目排行榜" id="顶级类目排行榜">顶级类目排行榜</a></h2>
<p>最后一个有意思的可视化是类目排行榜，从而了解哪些类目是支柱类目。不过由于源数据中的类目分类太细（约5000个类目），对于排行榜意义不大，因此我们希望能将其归约到顶级类目。所以笔者在 mysql 容器中预先准备了子类目与顶级类目的映射数据，用作维表。</p>
<p>在 SQL CLI 中创建 MySQL 表，后续用作维表查询。</p>
<pre><code class="language-sql">CREATE TABLE category_dim (
                           sub_category_id BIGINT,
                           parent_category_name STRING
                           ) WITH (
                           'connector' = 'jdbc',
                           'url' = 'jdbc:mysql://mysql:3306/flink',
                           'table-name' = 'category',
                           'username' = 'root',
                           'password' = '123456',
                           'lookup.cache.max-rows' = '5000',
                           'lookup.cache.ttl' = '10min'
                           );
</code></pre>
<p>同时我们再创建一个 Elasticsearch 表，用于存储类目统计结果。</p>
<pre><code class="language-sql">CREATE TABLE top_category (
                           category_name STRING PRIMARY KEY NOT ENFORCED,
                           buy_cnt BIGINT
                           ) WITH (
                           'connector' = 'elasticsearch-7',
                           'hosts' = 'http://elasticsearch:9200',
                           'index' = 'top_category'
                           );
</code></pre>
<p>第一步我们通过维表关联，补全类目名称。我们仍然使用 CREATE VIEW 将该查询注册成一个视图，简化逻辑。维表关联使用 temporal join 语法，可以查看文档了解更多：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/streaming/joins.html#join-with-a-temporal-table</p>
<pre><code class="language-sql">CREATE VIEW rich_user_behavior AS
SELECT U.user_id, U.item_id, U.behavior, C.parent_category_name as category_name
FROM user_behavior AS U LEFT JOIN category_dim FOR SYSTEM_TIME AS OF U.proctime AS C
ON U.category_id = C.sub_category_id;
</code></pre>
<p>最后根据 类目名称分组，统计出 <code>buy</code> 的事件数，并写入 Elasticsearch 中。</p>
<pre><code class="language-sql">INSERT INTO top_category
SELECT category_name, COUNT(*) buy_cnt
FROM rich_user_behavior
WHERE behavior = 'buy'
GROUP BY category_name;
</code></pre>
<p>提交上述查询后，在 Kibana 中创建 <code>top_category</code> 的 index pattern，然后在 Dashboard 中创建一个&quot;Horizontal Bar&quot;条形图，选择 <code>top_category</code> 索引，按照如下截图中的配置（左侧）画出类目排行榜，并保存。</p>
<p><img src="https://img.alicdn.com/tfs/TB13HW9weL2gK0jSZPhXXahvXXa-2874-1596.png" alt="" /></p>
<p>可以看到服饰鞋包的成交量远远领先其他类目。</p>
<p>Kibana 还提供了非常丰富的图形和可视化选项，感兴趣的用户可以用 Flink SQL 对数据进行更多维度的分析，并使用 Kibana 展示出可视化图，并观测图形数据的实时变化。</p>
<h2><a class="header" href="#结尾" id="结尾">结尾</a></h2>
<p>在本文中，我们展示了如何使用 Flink SQL 集成 Kafka, MySQL, Elasticsearch 以及 Kibana 来快速搭建一个实时分析应用。整个过程无需一行 Java/Scala 代码，使用 SQL 纯文本即可完成。期望通过本文，可以让读者了解到 Flink SQL 的易用和强大，包括轻松连接各种外部系统、对事件时间和乱序数据处理的原生支持、维表关联、丰富的内置函数等等。希望你能喜欢我们的实战演练，并从中获得乐趣和知识！</p>

                    </main>

                    <nav class="nav-wrapper" aria-label="Page navigation">
                        <!-- Mobile navigation buttons -->
                        
                            <a rel="prev" href="chapter13-10-00-使用Flink-SQL实现实时热门商品统计.html" class="mobile-nav-chapters previous" title="Previous chapter" aria-label="Previous chapter" aria-keyshortcuts="Left">
                                <i class="fa fa-angle-left"></i>
                            </a>
                        

                        
                            <a rel="next" href="chapter14-00-第十四章，常见面试题解答.html" class="mobile-nav-chapters next" title="Next chapter" aria-label="Next chapter" aria-keyshortcuts="Right">
                                <i class="fa fa-angle-right"></i>
                            </a>
                        

                        <div style="clear: both"></div>
                    </nav>
                </div>
            </div>

            <nav class="nav-wide-wrapper" aria-label="Page navigation">
                
                    <a rel="prev" href="chapter13-10-00-使用Flink-SQL实现实时热门商品统计.html" class="nav-chapters previous" title="Previous chapter" aria-label="Previous chapter" aria-keyshortcuts="Left">
                        <i class="fa fa-angle-left"></i>
                    </a>
                

                
                    <a rel="next" href="chapter14-00-第十四章，常见面试题解答.html" class="nav-chapters next" title="Next chapter" aria-label="Next chapter" aria-keyshortcuts="Right">
                        <i class="fa fa-angle-right"></i>
                    </a>
                
            </nav>

        </div>

        

        

        

        
        <script type="text/javascript">
            window.playground_copyable = true;
        </script>
        

        

        
        <script src="elasticlunr.min.js" type="text/javascript" charset="utf-8"></script>
        <script src="mark.min.js" type="text/javascript" charset="utf-8"></script>
        <script src="searcher.js" type="text/javascript" charset="utf-8"></script>
        

        <script src="clipboard.min.js" type="text/javascript" charset="utf-8"></script>
        <script src="highlight.js" type="text/javascript" charset="utf-8"></script>
        <script src="book.js" type="text/javascript" charset="utf-8"></script>

        <!-- Custom JS scripts -->
        

        

    </body>
</html>
